package com.gzc.just.play.server;

import com.gzc.just.play.Const;
import com.gzc.just.play.Utils;
import com.gzc.just.play.proto.War3Proto;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.Arrays;
import java.util.Collections;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

// 客户端对象，负责处理自身的消息和广播
class ClientSession implements Runnable {
    private static final Logger logger = LogManager.getLogger(ClientSession.class);
    
    private final String war3Ip;
    private final AsynchronousSocketChannel channel;
    private final LinkedBlockingQueue<AsynchronousSocketChannel> queue = new LinkedBlockingQueue<>();
    private final ReentrantLock lockObj = new ReentrantLock();

    public ClientSession(String war3Ip, AsynchronousSocketChannel channel) {
        this.war3Ip = war3Ip;
        this.channel = channel;
    }

    public String getWar3Ip() {
        return war3Ip;
    }

    public void putToQueue(AsynchronousSocketChannel channel) throws InterruptedException {
        queue.put(channel);
    }
    
    public AsynchronousSocketChannel pollQueue() throws InterruptedException {
        return queue.poll(3, TimeUnit.SECONDS);
    }

    // 通知客户端有新事件
    public void notification() {
        if (null == channel) {
            logger.error("connect is null,can not send data");
            return;
        }
        if (!channel.isOpen()) {
            logger.error("connect closed,can not send data");
            return;
        }

        try {
            War3Proto.Notification notification = War3Proto.Notification.newBuilder().build();
            War3Proto.War3Message message = War3Proto.War3Message.newBuilder()
                    .setType(War3Proto.MessageType.NOTIFICATION)
                    .setPayload(notification.toByteString())
                    .build();
            Utils.sendProtobufMessage(channel, message);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    // 更新客户端列表，有新的客户端加入
    public void updateClientList() {
        boolean lock = false;
        try {
            lock = lockObj.tryLock(Const.TIMEOUT_SECOND, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            logger.error(e.getMessage(), e);
        }
        if (!lock) {
            logger.error("updateClientList tryLock timeout fail,war3Ip={}", war3Ip);
            return;
        }

        try {
            if (null == channel) {
                logger.error("connect is null,can not send data");
                return;
            }
            if (!channel.isOpen()) {
                logger.error("connect closed,can not send data");
                return;
            }

            try {
                String result = Collections.list(Server.connectionMap.keys()).stream()
                        .filter(key -> !key.equals(war3Ip))  // 排除自身IP
                        .map(Object::toString)
                        .collect(Collectors.joining("#"));
                if (!result.isEmpty()) {
                    War3Proto.ClientListUpdate clientListUpdate = War3Proto.ClientListUpdate.newBuilder()
                            .addAllClientIps(Arrays.asList(result.split("#")))
                            .build();
                    War3Proto.War3Message message = War3Proto.War3Message.newBuilder()
                            .setType(War3Proto.MessageType.CLIENT_LIST_UPDATE)
                            .setPayload(clientListUpdate.toByteString())
                            .build();
                    logger.info("updateClientList,result={}", result);
                    Utils.sendProtobufMessage(channel, message);
                }
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        } catch (Exception e) {
            logger.error("updateClientList error", e);
        } finally {
            lockObj.unlock();
        }
    }

    // 监听客户端消息并进行广播
    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                ByteBuffer lengthBuffer = ByteBuffer.allocate(4);
                int bytesRead = channel.read(lengthBuffer).get();
                if (-1 == bytesRead) {
                    logger.error("client disconnected,RemoteAddress={}", channel.getRemoteAddress());
                    break;
                }

                lengthBuffer.flip();
                int messageLength = lengthBuffer.getInt();
                ByteBuffer messageBuffer = ByteBuffer.allocate(messageLength);
                channel.read(messageBuffer).get();
                War3Proto.War3Message message = War3Proto.War3Message.parseFrom(messageBuffer.array());

                if (message.getType() == War3Proto.MessageType.DATA_FORWARD) {
                    War3Proto.DataForward dataForward = War3Proto.DataForward.parseFrom(message.getPayload());
                    String otherWar3Ip = dataForward.getFromIp();
                    String dstIp = message.getTargetIp();
                    ClientSession clientSession = Server.connectionMap.get(otherWar3Ip);
                    logger.info("forward data,dstIp={},otherWar3Ip={}", dstIp, otherWar3Ip);

                    if (clientSession != null && clientSession.channel != null && clientSession.channel.isOpen()) {
                        War3Proto.War3Message forwardMessage = War3Proto.War3Message.newBuilder()
                                .setType(War3Proto.MessageType.DATA_FORWARD)
                                .setSourceIp(otherWar3Ip)
                                .setTargetIp(dstIp)
                                .setPayload(dataForward.toByteString())
                                .build();
                        Utils.sendProtobufMessage(clientSession.channel, forwardMessage);
                    }
                }
            }
        } catch (Exception e) {
            logger.error("client run error", e);
        } finally {
            try {
                if (null != channel) {
                    channel.close();
                }
                // 从连接映射中移除自身
                Server.connectionMap.remove(war3Ip);
                // 释放ip
                Server.ipPool.releaseIP(war3Ip);
            } catch (IOException e) {
                logger.error(e.getMessage(), e);
            }
        }
    }
}
